package com.atguigu.chapter07;

import com.atguigu.chapter05.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/6 9:12
 */
public class Flink07_TimeWindow_SourceCodeAnalysis {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        // 切分
                        String[] line = value.split(",");
                        return new WaterSensor(line[0], Long.valueOf(line[1]), Integer.valueOf(line[2]));

                    }
                });



        KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(sensor -> sensor.getId());

        // 分组之后开窗：
        WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorKS
                .window(TumblingEventTimeWindows.of(Time.seconds(5)));


        sensorWS
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                    @Override
                    public void process(String key, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                        out.collect("key=" + key + "\n" +
                                "数据条数:" + elements.spliterator().estimateSize() + "\n" +
                                "窗口是:[" + context.window().getStart() + "," + context.window().getEnd() + ")\n\n");
                    }
                })
                .print();


        env.execute();
    }
}
/*
    以 基于 事件时间的 滚动窗口（10s） 为例：

        1、窗口是怎么划分的？为什么是左闭右开？
            1）start = timestamp - (timestamp - offset + windowSize) % windowSize;
                    事件时间 - （事件时间 - 0 + 窗口大小）%窗口大小
                    3 - （3+10）%10 = 0
                    14 - （14 + 10）%10 = 10

                时间戳按照窗口长度 取整数倍（以1970年1月1日0点为起点 => 伦敦时间）
            2）end = start + size
                开始时间 + 窗口长度

            3）左闭右开：   属于本窗口的最大时间戳 = end -1ms  ， 所以时间为 end这条数据，不属于本窗口，所以是开区间



        2、窗口什么时候触发、输出？
             window.maxTimestamp() <= ctx.getCurrentWatermark()
             当前的 watermark >= 窗口的最大时间戳


        3、窗口的生命周期：什么创建的？

            Collections.singletonList(new TimeWindow(start, start + size));
            本窗口（都是同一分组）的第一条数据来的时候，使用 new的方式创建的，添加到一个单例的集合


        4、窗口的生命周期：什么时候关闭的？
                cleanupTime = window.maxTimestamp() + allowedLateness;
                allowedLateness 默认是 0ms
              当时间（watermark）达到 窗口的最大时间戳 + 允许迟到的时间，就会关闭窗口
 */